package com.zmj.tl.database.taos.listener;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.zmj.tl.database.taos.highvolume.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author 033529
 * @version 1.0
 * @title IotKafkaListener
 * @description 监听kafka消息，将数据写入到td时序数据库中
 * @create 2024/5/4 15:36
 */
@Component
public class IotKafkaListener {
    private final Logger logger = LoggerFactory.getLogger(IotKafkaListener.class);

    //final static int TASK_QUEUE_CAPACITY = 1000000;
    final static int TASK_QUEUE_CAPACITY = 1000;
    final static List<BlockingQueue<String>> BLOCKING_QUEUES = new ArrayList<>();
    final static List<ReadTask> READ_TASKS = new ArrayList<>();
    final static List<WriteTask> WRITE_TASKS = new ArrayList<>();
    final static DataBaseMonitor DATA_BASE_MONITOR = new DataBaseMonitor();

    @KafkaListener(topics = {"iot.data"}, groupId = "iot.consumer.data.list")
    public void handler(List<org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>> consumerRecords, Acknowledgment ack) {
        if (!CollectionUtils.isEmpty(consumerRecords)) {
            logger.info("接受的消息:[{}]", consumerRecords);
            for (ConsumerRecord<?, ?> consumerRecord : consumerRecords) {
                String value = (String) consumerRecord.value();
                List<Map<String, Object>> lists = JSON.parseObject(JSONUtil.toJsonStr(value), new TypeReference<List<Map<String, Object>>>() {
                });
                for (Map<String, Object> list : lists) {
                    try {
                        String s = fastWrite(list);
                        logger.info("写入结果：" + s);
                    } catch (Exception e) {
                        logger.error("快速写入数据失败:", e);
                        //throw new RuntimeException(e);
                    }
                }
            }
        }
        ack.acknowledge();
    }


    public static void stopAll() {
        READ_TASKS.forEach(ReadTask::stop);
        WRITE_TASKS.forEach(WriteTask::stop);
        DATA_BASE_MONITOR.close();
    }


    private String fastWrite(Map<String, Object> map) throws Exception {
        int readTaskCount = 1;
        int writeTaskCount = 3;
        int tableCount = 1000;
        int maxBatchSize = 3000;
        ExecutorService pool = Executors.newFixedThreadPool(100);

        logger.info("readTaskCount={}, writeTaskCount={} tableCount={} maxBatchSize={}",
                readTaskCount, writeTaskCount, tableCount, maxBatchSize);
        DATA_BASE_MONITOR.init();

        // Create task queues, whiting tasks and start writing threads.
//        for (int i = 0; i < writeTaskCount; ++i) {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(TASK_QUEUE_CAPACITY);
        BLOCKING_QUEUES.add(queue);
//        WriteTask task = new WriteTask(queue, maxBatchSize);

        pool.submit(new Runnable() {
            @Override
            public void run() {
                logger.info("started");
                String line = null; // data getting from the queue just now.
                SQLWriter writer = new SQLWriter(maxBatchSize);
                try {
                    writer.init();
                    line = queue.poll();
                    if (line != null) {
                        // parse raw data and buffer the data.
                        writer.processLine(line);
                    }
//            while (active) {
//                line = queue.poll();
//                if (line != null) {
//                    // parse raw data and buffer the data.
//                    writer.processLine(line);
//                }
////                else if (writer.hasBufferedValues()) {
////                    // write data immediately if no more data in the queue
////                    writer.flush();
////                } else {
////                    // sleep a while to avoid high CPU usage if no more data in the queue and no buffered records, .
////                    Thread.sleep(100);
////                }
//            }
//            if (writer.hasBufferedValues()) {
//                writer.flush();
//            }
                } catch (Exception e) {
                    String msg = String.format("line=%s, bufferedCount=%s", line, writer.getBufferedCount());
                    logger.error(msg, e);
                } finally {
//                        try {
//                            writer.close();
//                        } catch (Exception e) {
//                            logger.error("writer 关闭失败", e);
//                        }

                }
            }
        });
//            Thread t = new Thread(task);
//            t.setName("WriteThread-" + i);
//            t.start();
//         }


        // create reading tasks and start reading threads
        int tableCountPerTask = tableCount / readTaskCount;
//        for (int i = 0; i < readTaskCount; ++i) {
        ReadTask task = new ReadTask(1, BLOCKING_QUEUES, tableCountPerTask, map);
        pool.submit(new Runnable() {
            @Override
            public void run() {
                logger.info("started");
                MockDataSource it = new MockDataSource("tb" + 1, tableCount, map);
                try {
                    while (it.hasNext()) {
                        String line = it.next();
                        int queueId = getQueueId(line);
                        BLOCKING_QUEUES.get(queueId).put(line);
                    }
                } catch (Exception e) {
                    logger.error("Read Task Error", e);
                }
            }
        });
//            Thread t = new Thread(task);
//            t.setName("ReadThread-" + i);
//            t.start();
//        }

        Runtime.getRuntime().addShutdownHook(new Thread(IotKafkaListener::stopAll));

//        long lastCount = 0;
//        while (true) {
//            Thread.sleep(10000);
//            long numberOfTable = DATA_BASE_MONITOR.getTableCount();
//            long count = DATA_BASE_MONITOR.count();
//            logger.info("numberOfTable={} count={} speed={}", numberOfTable, count, (count - lastCount) / 10);
//            lastCount = count;
//        }
        return "success";

    }

    public int getQueueId(String line) {
        String tbName = line.substring(0, line.indexOf(',')); // For example: tb1_101
        String suffixNumber = tbName.split("_")[1];
        return Integer.parseInt(suffixNumber) % BLOCKING_QUEUES.size();
    }

}
